package com.atguigu.edu.realtime.common.util;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.constant.Constant;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.AsyncConnection;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> {
    private StatefulRedisConnection<String, String> redisConnection;
    private AsyncConnection hbaseConn;

    @Override
    public void open(Configuration parameters) throws Exception {
        redisConnection = RedisUtil.getRedisAsyncConnection();
        hbaseConn = HBaseUtil.getHBaseAsyncConn();
    }

    @Override
    public void close() throws Exception {
        RedisUtil.closeRedisAsyncConnection(redisConnection);
        HBaseUtil.closeHBaseAsyncConn(hbaseConn);
    }
    @Override
    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) throws Exception {
        //创建异步编排对象,有返回值，作为下一个线程任务参数
        CompletableFuture.supplyAsync(
                new Supplier<JSONObject>() {
                    @Override
                    public JSONObject get() {
                        //根据主键以异步的方式从redis获取维度数据
                        JSONObject dimJsonObj = RedisUtil.readDimAsync(redisConnection, getTableName(), getRowKey(obj));

                        return dimJsonObj;
                    }
                }
        ).thenApplyAsync(
                //创建线程任务，有入参，有返回值
                new Function<JSONObject, JSONObject>() {
                    @Override
                    public JSONObject apply(JSONObject dimJsonObj) {
                        //缓存命中：?去hbase找
                        if (dimJsonObj!=null){
                            System.out.println("Redis中获取"+getTableName()+"表的"+getRowKey(obj)+"数据");
                        }else {
                            dimJsonObj=HBaseUtil.readDimAsync(hbaseConn, Constant.HBASE_NAMESPACE,getTableName(),getRowKey(obj));
                            if (dimJsonObj!=null){
                                System.out.println("HBase中获取"+getTableName()+"表的"+getRowKey(obj)+"数据");
                                RedisUtil.writeDimAsync(redisConnection,getTableName(),getRowKey(obj),dimJsonObj);
                            }else {
                                System.out.println("无法从HBase中读到"+getTableName()+"表的"+getRowKey(obj)+"数据");
                            }
                        }
                        //将hbase中结果写到redis
                        return dimJsonObj;
                    }
                }
        ).thenAccept(
                //有入参，无返回值
                new Consumer<JSONObject>() {
                    @Override
                    public void accept(JSONObject dimJsonObj) {
                        if (dimJsonObj!=null){
                            addDim(obj,dimJsonObj);
                        }
                        //获取数据库交互的结果并发送给resultFuture的回调函数（向下游传递数据）
                        resultFuture.complete(Collections.singleton(obj));

                    }
                }
        );
    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        throw new RuntimeException("检查进程，维度表历史数据是否初始化");
    }

    protected abstract void addDim(T obj, JSONObject dimJsonObj);

    public abstract String getTableName() ;

    public abstract String getRowKey(T obj) ;
}

